#include <iostream>
#include <mutex>
#include <thread>
#include <memory>

#include <engine.h>
#include <matrix.h>

#include <boost/locale.hpp>
#include <boost/dll/shared_library.hpp>
#include <boost/filesystem.hpp>
#include <boost/exception/all.hpp>
#include <boost/process.hpp>
#include <boost/stacktrace.hpp>
#include <boost/asio.hpp>

#include "gen-cpp/CallService.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TServerFramework.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>

using namespace std;
using namespace boost;
using namespace apache::thrift::server;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;
#define MATLABSIZE 1024

Engine* engine=nullptr;
atomic<bool> is_exit(false);
condition_variable cbegin_value,cend_value,rbegin_value,rend_value;
mutex task_mutex,recv_mutex,return_mutex;
deque<std::tuple<string,string,string>> tasks;
class CallServiceHandler : virtual public CallServiceIf {
public:
    CallServiceHandler() {}
    void Process(std::string& _return, const std::string& file, const std::string& value) {
        {
            lock_guard<mutex> lock(task_mutex);
            //tasks.emplace_back(make_tuple(boost::locale::conv::between(file,"utf8","gbk"),boost::locale::conv::between(value,"utf8","gbk"),string()));
            tasks.emplace_back(make_tuple(file,value,string()));
        }

        unique_lock<mutex> r_lock(return_mutex);
        cbegin_value.notify_all();
        rbegin_value.wait(r_lock);
        {
            lock_guard<mutex> lock(task_mutex);
            _return = get<2>(*tasks.begin());
            tasks.erase(tasks.begin());
        }
        cend_value.notify_all();
        rend_value.wait(r_lock);
    }
};
class MultiServer : public TServerFramework {
public:
  MultiServer(
      const std::shared_ptr<apache::thrift::TProcessorFactory>& processorFactory,
      const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory):
        TServerFramework(processorFactory, serverTransport, transportFactory, protocolFactory){
      }

  MultiServer(
      const std::shared_ptr<apache::thrift::TProcessor>& processor,
      const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory):
        TServerFramework(processor, serverTransport, transportFactory, protocolFactory){
      }
  MultiServer(
      const std::shared_ptr<apache::thrift::TProcessorFactory>& processorFactory,
      const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& inputTransportFactory,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& outputTransportFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& inputProtocolFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& outputProtocolFactory):
        TServerFramework(processorFactory, serverTransport, inputTransportFactory, outputTransportFactory,inputProtocolFactory,outputProtocolFactory){
      }

  MultiServer(
      const std::shared_ptr<apache::thrift::TProcessor>& processor,
      const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& inputTransportFactory,
      const std::shared_ptr<apache::thrift::transport::TTransportFactory>& outputTransportFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& inputProtocolFactory,
      const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& outputProtocolFactory):
        TServerFramework(processor, serverTransport, inputTransportFactory, outputTransportFactory, inputProtocolFactory, outputProtocolFactory){
      }

  ~MultiServer() override{};

protected:
  void onClientConnected(const std::shared_ptr<TConnectedClient>& pClient) override {
    cout<<"connect servers count,"<<getConcurrentClientCount()<<endl;
    pClient->run();
  }
  void onClientDisconnected(TConnectedClient* pClient) override {
  }
};
int main(int argc, char* argv[]){
    int port = argc>1?atoi(argv[1]):0;
    if(port==0){
        cout<<"port parameter error."<<endl;
        return 0;
    }
    boost::system::error_code ec;
    boost::asio::io_service io_serv;
    boost::asio::ip::tcp::socket port_test(io_serv);
    port_test.open(boost::asio::ip::tcp::v4(), ec);
    port_test.bind(boost::asio::ip::tcp::endpoint(boost::asio::ip::address::from_string("127.0.0.1"), port) , ec);
    if(ec == boost::asio::error::address_in_use || ec == boost::asio::error::access_denied){
        cout<<"process is already running."<<endl;
        return 0;
    }
    port_test.close(ec);

    engine=engOpen(0);
    engSetVisible(engine,false);
    std::cout<<"---------------"<<endl<<"server start"<<endl<<"---------------"<<endl;
    thread([&]{
        try {        
            std::shared_ptr<CallServiceHandler> handler(new CallServiceHandler());
            std::shared_ptr<TProcessor> processor(new CallServiceProcessor(handler));
            std::shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
            std::shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
            std::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

            MultiServer server(processor, serverTransport, transportFactory, protocolFactory);
            server.setConcurrentClientLimit(10);
            server.serve();
        }catch (boost::exception &e) {
            cout << boost::stacktrace::stacktrace() << endl;
            cout << boost::diagnostic_information(e) << endl;
        }
    }).detach();
    while(!is_exit){
        unique_lock<mutex> c_lock(recv_mutex);
        try {
            stringstream ret;
            rend_value.notify_all();
            cbegin_value.wait(c_lock);
            string file_path,value;
            {
                lock_guard<mutex> lock(task_mutex);
                if(tasks.size()==0) continue;
                file_path = get<0>(tasks.front());
                value = get<1>(tasks.front());
            }
            ret<<"------------------------------"<<endl<<
                "filename:"<<file_path<<endl<<"parameter:"<<value<<endl<<"output:";
            boost::filesystem::path filepath(file_path);
            string extension = filepath.extension().string();
            if(extension == ".m"){
                char textOut[MATLABSIZE] = {};
                engOutputBuffer(engine,textOut,sizeof(textOut));
                engEvalString(engine,("addpath('"+boost::filesystem::absolute(filepath).parent_path().string()+"')").c_str());
                if(value.empty()){
                    engEvalString(engine,("run(\""+filepath.string()+"\")").c_str());
                }else{
                    engEvalString(engine,(filepath.stem().string()+"(jsondecode('"+value+"'))").c_str());
                }
                engOutputBuffer(engine, NULL, 0);
                ret <<endl<< textOut << endl;
                engEvalString(engine,"jsonencode(ans)");
                mxArray* result = engGetVariable(engine,"ans");
                if(result!=nullptr){
                    char* p = mxArrayToString(result);
                    if(p) {
                        {
                            lock_guard<mutex> lock(task_mutex);
                            get<2>(tasks.front()) = p;
                        }
                        mxDestroyArray(result);
                        rbegin_value.notify_all();
                        cend_value.wait(c_lock);
                        continue;
                    }
                }else{
                    ret<<endl;
                }
                mxDestroyArray(result);
            }else{
                ret <<"file extension eror"<<endl;
            }
            ret<<"------------------------------"<<endl;
            cout << ret.str() << flush; 
            {
                lock_guard<mutex> lock(task_mutex);
                get<2>(tasks.front()) = ret.str();
            }
            rbegin_value.notify_all();
            cend_value.wait(c_lock);
        }catch (boost::exception &e) {
            rbegin_value.notify_all();
            cend_value.wait(c_lock);

            cout << boost::stacktrace::stacktrace() << endl;
            cout << boost::diagnostic_information(e) << endl;
        }
    }
    if(engine) engClose(engine);
    return 0;
}